package cn.datawin.cmppserver.init;

import cn.datawin.bms.customer.redis.CstmCustomerCmppRedisDto;
import cn.datawin.cmppserver.netway.handler.CmppServerCustomHandler;
import cn.datawin.cmppserver.netway.provider.TcpMessageProvider;
import cn.datawin.cmppserver.netway.validator.TcpValidator;
import cn.datawin.cmppserver.queue.QueueCache;
import cn.datawin.cmppserver.util.CacheService;
import com.alibaba.fastjson.JSONObject;
import com.drondea.sms.channel.ChannelSession;
import com.drondea.sms.conf.cmpp.CmppServerSocketConfig;
import com.drondea.sms.session.AbstractServerSessionManager;
import com.drondea.sms.session.cmpp.CmppServerSessionManager;
import com.drondea.sms.type.CmppConstants;
import com.drondea.sms.type.DefaultEventGroupFactory;
import io.netty.channel.Channel;
import io.netty.util.ResourceLeakDetector;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;


/**
 * 初始化系统环境
 */
@Slf4j
@Component
public class InitSystemEnv implements CommandLineRunner {



    public static final String CMPP_SERVER_ID = "cmpp_server";
    @Resource
    private CacheService cacheService;

    private CmppServerSessionManager cmppServerSessionManager;

    @Override
    public void run(String... args) {
        try {
            log.info("cmpp服务端初始化.....");
            List<CstmCustomerCmppRedisDto> customerCmppRedisDtoList = cacheService.getAllValidCustomer();
            log.info("customerCmppRedisDtoList={}", JSONObject.toJSONString(customerCmppRedisDtoList));
			log.info("-------------------------------------------------");
            for (CstmCustomerCmppRedisDto customer : customerCmppRedisDtoList) {
                if (customer.getStatus() == 1) {
                    String customerCode = customer.getCustomerCode();
                    QueueCache.MESSAGES_REPORT_QUEUE.putIfAbsent(customerCode, new ConcurrentLinkedQueue<>());
                    QueueCache.MESSAGES_INBOX_QUEUE.putIfAbsent(customerCode, new ConcurrentLinkedQueue<>());
                }
            }

            cmppServerSessionManager = startSessionManager();
        } catch (Exception e) {
            log.error("cmpp服务端启动失败", e);
        }


    }


    public CmppServerSessionManager startSessionManager() {
        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);

        // 获取配置端口，为空获取CMPP默认端口
        int port = 7891;
        CmppServerSocketConfig socketConfig = new CmppServerSocketConfig(CMPP_SERVER_ID, port);

        //服务器端默认版本号2.0
        socketConfig.setVersion(CmppConstants.VERSION_20);

        CmppServerCustomHandler customHandler = new CmppServerCustomHandler();

        CmppServerSessionManager sessionManager = new CmppServerSessionManager(new TcpValidator(), socketConfig, customHandler);
        sessionManager.setMessageProvider(new TcpMessageProvider());
        sessionManager.doOpen();

        log.debug("启动CMPP网关，端口:" + port);
        return sessionManager;
    }

    /**
     * 用户没有连接后关闭相关的资源
     * 60秒钟检测一次，如果没有连接就关闭consumer
     */
    @Scheduled(fixedDelay = 10 * 1000)
    private void closeUserResource() {
        List<CstmCustomerCmppRedisDto> allLoginUser = cacheService.getAllValidCustomer();
        allLoginUser.forEach(user -> {
            updateUserTcpInfo(cmppServerSessionManager, user);
        });
    }

    private void updateUserTcpInfo(AbstractServerSessionManager abstractServerSessionManager,
                                   CstmCustomerCmppRedisDto user) {
        if (abstractServerSessionManager == null) {
            return;
        }
        String account = user.getCustomerCode();
        int userSessionSize = abstractServerSessionManager.getUserSessionSize(account);
        log.debug("account={},sessions={}", account, userSessionSize);

        if (userSessionSize == 0 || user.getStatus().equals(0)) {
            abstractServerSessionManager.closeUserSession(account);
            cacheService.updateConnections(user.getCustomerId(), "");
        } else {
            //修改用户的提交速度
            modifyUserSpeed(abstractServerSessionManager, user);
            Map<String, Long> connectInfo = countConnection(abstractServerSessionManager, account);
            cacheService.updateConnections(user.getCustomerId(), connectInfo == null ? "" : JSONObject.toJSONString(connectInfo));
        }

    }

    private Map<String, Long> countConnection(AbstractServerSessionManager abstractServerSessionManager, String tcpUserName) {
        if (abstractServerSessionManager == null) {
            return null;
        }
        List<ChannelSession> userSessions = abstractServerSessionManager.getUserSessions(tcpUserName);
        if (userSessions == null) {
            return null;
        }

        Map<String, Long> userIpCounting = userSessions.stream().collect(Collectors.groupingBy(userSession -> {
            Channel channel = userSession.getChannel();
            if (channel == null) {
                return "unknown";
            }
            return ((InetSocketAddress) channel.remoteAddress())
                    .getAddress().getHostAddress();
        }, Collectors.counting()));

        return userIpCounting;
    }

    private void modifyUserSpeed(AbstractServerSessionManager abstractServerSessionManager, CstmCustomerCmppRedisDto user) {
        int submitSpeed = 200;
        if (user.getAcceptRate() != null) {
            submitSpeed = user.getAcceptRate();
        }
        //修改用户提交速度
        abstractServerSessionManager.resetUserSubmitSpeed(user.getCustomerCode(), submitSpeed);
    }



    @PreDestroy
    public void release() throws InterruptedException {
        log.debug("关闭netway，销毁资源=====");
        log.debug("关闭拉取程序。");
        Thread.sleep(3000);
        log.debug("关闭tcp连接。");
        if (cmppServerSessionManager != null) {
            cmppServerSessionManager.doClose();
        }

        DefaultEventGroupFactory.getInstance().close();

        Runtime.getRuntime().exit(0);
    }
}
